package com.test.mqtt.client;

import com.test.mqtt.QosEnum;
import com.test.mqtt.properties.EmqxProperties;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
@Component
public class EmqClient {
    private static final Logger log = LoggerFactory.getLogger(EmqClient.class);

    private IMqttClient iMqttClient;
    @Autowired
    private EmqxProperties emqxProperties;

    @Autowired
    private MqttCallback mqttCallback;
    // 初始化
    @PostConstruct
    public void init(){
        MqttClientPersistence memoryPersistence = new MemoryPersistence();
        try {
            iMqttClient = new MqttClient(emqxProperties.getBrokerUrl(), emqxProperties.getClientId(),memoryPersistence);
        } catch (MqttException e) {
            //e.printStackTrace();
            log.error("客户端初始化失败");
        }
    }

    /**
     * 订阅主题
     * @param topicFilters  主题通配符
     * @param qos       消息级别
     */
    public void subscribe(String topicFilters,QosEnum qos){
        try {
            iMqttClient.subscribe(topicFilters,qos.value());
        } catch (MqttException e) {
            //e.printStackTrace();
            log.error("订阅主题失败，错误原因为：" + e.getMessage());
        }
    }

    /**
     * 取消订阅
     * @param topicFilters 主题
     */
    public void unSubscribe(String topicFilters){
        try {
            iMqttClient.unsubscribe(topicFilters);
        } catch (MqttException e) {
            //e.printStackTrace();
            log.error("取消订阅主题失败，失败原因为：" + e.getMessage());
        }
    }

    /**
     * 发布消息
     * @param topic 主题
     * @param msg 消息内容
     * @param qos 消息级别枚举
     * @param retain 是否是保留消息
     */
    public void publish(String topic, String msg, QosEnum qos, boolean retain ){
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setPayload(msg.getBytes(StandardCharsets.UTF_8));
        mqttMessage.setQos(qos.value());
        mqttMessage.setRetained(retain);
        try {
            iMqttClient.publish(topic,mqttMessage);
        } catch (MqttException e) {
            //e.printStackTrace();
            log.error("消息发布失败，失败原因：" + e.getMessage());
        }
    }

    /**
     * 连接服务器
     * @param username
     * @param password
     */
    public void connect(String username,String password){
        MqttConnectOptions connectOptions = new MqttConnectOptions();
        // 自动尝试连接服务端
        connectOptions.setAutomaticReconnect(true);
        connectOptions.setUserName(username);
        connectOptions.setPassword(password.toCharArray());
        // 设置为 false 代表是客户端和服务端保持常连接，客户端断开，服务端的会话也会保持，知道到达持续时间
        // 设置为 true 代表临时会话
        connectOptions.setCleanSession(true); //临时会话
        // 设置方法回调
        // 客户端收到消息之后，就需要执行callback
        iMqttClient.setCallback(mqttCallback);

        try {
            iMqttClient.connect(connectOptions);
        } catch (MqttException e) {
            //e.printStackTrace();
            log.error("mqtt客户端连接服务端失败，失败原因是：" + e.getMessage());
        }
    }

    /**
     * 重新连接
     */
    public void reConnect(){
        try {
            iMqttClient.reconnect();
        } catch (MqttException e) {
            //e.printStackTrace();
            log.error("mqtt客户端重新连接失败，失败原因是：" + e.getMessage());
        }
    }

    /**
     * 断开连接
     */
    @PreDestroy  //服务结束前自动断开连接
    public void disConnect(){
        try {
            iMqttClient.disconnect();
        } catch (MqttException e) {
            //e.printStackTrace();
            log.error("mqtt客户端断开连接失败，失败原因是" + e.getMessage());
        }
    }
}
